﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
using Orleans.Streams;
using Vertex.Abstractions.Actor;
using Vertex.Abstractions.Event;
using Vertex.Abstractions.Serialization;
using Vertex.Abstractions.Snapshot;
using Vertex.Abstractions.Storage;
using Vertex.Protocol;
using Vertex.Runtime.Actor;
using Vertex.Runtime.Event;
using Vertex.Runtime.Exceptions;
using Vertex.Runtime.Options;
using Vertext.Abstractions.Event;
using Walterlv.WeakEvents;
using System.Reactive.Linq;
using System.Reactive.Disposables;
using System.Reactive.Subjects;

namespace Vertex.Storage.Linq2db
{
    /// <summary>
    /// 流的观察者
    /// </summary>
    /// <typeparam name="T"></typeparam>
    public interface IInitObserver<in T, TPrimaryKey> : IAsyncObserver<T>, IAsyncDisposable
    {
        //event ReceiveEventHandler OnReciiveEvent;
        Subject<IEvent> EventSubject { get; }
        ValueTask Init();

        object GetState();

        Task<IList<EventUnit<TPrimaryKey>>> GetEventList(long startVersion, long count);

        /// <summary>
        /// 查询某个 时间戳节点前快照状态
        /// </summary>
        /// <param name="timestamp">时间戳</param>
        /// <returns></returns>
        Task<object> GetSnapshotByTimestamp(long timestamp);
    }

    /// <summary>
    /// 负责快照的 读取 存储，事件的查询
    /// </summary>
    /// <typeparam name="TPrimaryKey">TPrimaryKey</typeparam>
    /// <typeparam name="T">T</typeparam>
    public abstract class SnapshotService<TPrimaryKey, T> : IInitObserver<byte[], TPrimaryKey>
          where T : ISnapshot, new()
    {
        private IActor<TPrimaryKey> actor;

        public SnapshotService(IActor<TPrimaryKey> actor, IServiceProvider serviceProvider)
        {
            this.ActorId = actor.ActorId;
            this.actor = actor;
            this.ServiceProvider = serviceProvider;
            this.ActorType = actor.GetType();

        }

        /// <summary>
        /// 事件流
        /// </summary>
        public Subject<IEvent> EventSubject { get; } = new Subject<IEvent>();

        /// <summary>
        /// 由于此方法 可能会触发 创建数据库表，因此 需要 外面手动调用 await，而不能在 构造函数中调用
        /// </summary>
        /// <returns></returns>
        protected virtual async ValueTask DependencyInjection()
        {
            this.VertexOptions = this.ServiceProvider.GetService<IOptionsMonitor<ActorOptions>>().Get(this.ActorType.FullName);
            this.ArchiveOptions = this.ServiceProvider.GetService<IOptionsMonitor<ArchiveOptions>>().Get(this.ActorType.FullName);

            this.Serializer = this.ServiceProvider.GetService<ISerializer>();
            this.EventTypeContainer = this.ServiceProvider.GetService<IEventTypeContainer>();
            this.SnapshotHandler = this.ServiceProvider.GetService<ISnapshotHandler<TPrimaryKey, T>>();
            if (this.SnapshotHandler == default)
            {
                throw new VertexEventHandlerException(this.ActorType);
            }

            var eventStorageFactory = this.ServiceProvider.GetService<IEventStorageFactory>();

            // 由于这里 需要去Acotr上面的特性 去查表名，因此 这里的Actor必须是实际 Actor
            this.EventStorage = await eventStorageFactory.Create(this.actor);
            var eventArchiveFactory = this.ServiceProvider.GetService<IEventArchiveFactory>();
            this.EventArchive = await eventArchiveFactory.Create(this.actor);
            var snapshotStorageFactory = this.ServiceProvider.GetService<ISnapshotStorageFactory>();
            this.SnapshotStorage = await snapshotStorageFactory.Create(this.actor);

            // 用完直接释放即可
            this.actor = null;
        }

        protected Type ActorType { get; set; }

        protected TPrimaryKey ActorId { get; set; }

        public IServiceProvider ServiceProvider { get; }

        protected IEventArchive<TPrimaryKey> EventArchive { get; private set; }

        protected ActorOptions VertexOptions { get; private set; }

        protected ArchiveOptions ArchiveOptions { get; private set; }

        protected ISnapshotHandler<TPrimaryKey, T> SnapshotHandler { get; private set; }

        /// <summary>
        /// Event storage
        /// </summary>
        protected IEventStorage<TPrimaryKey> EventStorage { get; private set; }

        /// <summary>
        /// State storage
        /// </summary>
        protected ISnapshotStorage<TPrimaryKey> SnapshotStorage { get; private set; }

        public SnapshotUnit<TPrimaryKey, T> Snapshot { get; set; }

        /// <summary>
        /// The event version number of the snapshot
        /// </summary>
        protected long ActivateSnapshotVersion { get; private set; }

        protected IEventTypeContainer EventTypeContainer { get; private set; }

        protected ISerializer Serializer { get; private set; }



        #region 弱事件相关
        ///// <summary>
        ///// 使用弱事件机制，当处理完事件时，触发此事件
        ///// </summary>
        //public event ReceiveEventHandler OnReciiveEvent
        //{
        //    add => this.receiveEvent.Add(value, value.Invoke);
        //    remove => this.receiveEvent.Remove(value);
        //}

        //public event ReceiveEventHandler OnReceiveEventBefore
        //{
        //    add => this.receiveEventBefore.Add(value, value.Invoke);
        //    remove => this.receiveEventBefore.Remove(value);
        //}

        ///// <summary>
        ///// 此事件是为了让组件强制刷新使用
        ///// </summary>

        //private readonly WeakEvent<IEvent> receiveEvent = new WeakEvent<IEvent>();
        ///// <summary>
        ///// 此事件是为让 全局共享的状态 更新使用
        ///// 因为全局共享状态 只需要一次 更新即可，而每个组件都需要 强制刷新
        ///// 因此就需要使用一个独立的事件，触发共享状态 更新
        ///// 然后再使用 receiveEvent事件，触发每个组件单独更新
        ///// 如果不使用全局共享状态，则无需使用此事件
        ///// </summary>
        //private readonly WeakEvent<IEvent> receiveEventBefore = new WeakEvent<IEvent>();

        //private void InvokeReceiveEvent(IEvent @event) => this.receiveEvent?.Invoke(this, @event);
        //private void InvokeReceiveEventBefore(IEvent @event) => this.receiveEventBefore?.Invoke(this, @event);
        #endregion

        /// <summary>
        /// Initialization state, must be implemented
        /// </summary>
        /// <returns></returns>
        protected virtual ValueTask CreateSnapshot()
        {
            this.Snapshot = new SnapshotUnit<TPrimaryKey, T>
            {
                Meta = new SnapshotMeta<TPrimaryKey>
                {
                    ActorId = this.ActorId,
                    MinEventTimestamp = DateTimeOffset.UtcNow.ToUnixTimeSeconds()
                },
                Data = new T()
            };
            return ValueTask.CompletedTask;
        }

        protected IList<EventUnit<TPrimaryKey>> Convert(IList<EventDocument<TPrimaryKey>> documents)
        {
            return documents.Select(document =>
            {
                if (!this.EventTypeContainer.TryGet(document.Name, out var type))
                {
                    throw new NoNullAllowedException($"event name of {document.Name}");
                }
                var data = this.Serializer.Deserialize(document.Data, type);
                return new EventUnit<TPrimaryKey>
                {
                    ActorId = this.ActorId,
                    Event = data as IEvent,
                    Meta = new EventMeta { Version = document.Version, Timestamp = document.Timestamp, FlowId = document.FlowId }
                };
            }).ToList();
        }

        public async Task<IList<EventUnit<TPrimaryKey>>> GetEventList(long startVersion, long count)
        {
            var documents = await this.EventStorage.GetList(this.ActorId, startVersion, startVersion + count);

            if (documents.Count == 0)
            {
                return Array.Empty<EventUnit<TPrimaryKey>>();
            }

            return this.Convert(documents);
        }

        /// <summary>
        /// 查询某一时间戳之前的对象状态，就是定版时的状态
        /// </summary>
        /// <returns></returns>
        public async Task<object> GetSnapshotByTimestamp(long timestamp)
        {
            var snapshot = new SnapshotUnit<TPrimaryKey, T>
            {
                Meta = new SnapshotMeta<TPrimaryKey>
                {
                    ActorId = this.ActorId,
                },
                Data = new T()
            };
            await this.CreateGetEventFromTimestamp(timestamp).ForEachAsync(unit =>
            {
                this.SnapshotHandler.Apply(snapshot, unit);
                snapshot.Meta.UpdateVersion(unit.Meta, this.ActorType); // Version of the update process
            });

            return snapshot.Data;
        }
        /// <summary>
        /// 恢复快照状态
        /// 有两种情况会调用此方发
        /// 1.首次启动初始化时，使用此方法恢复状态后，开启事件流监听
        /// 2.在应用事件的过程中，发现快照版本于事件版本不一致，则重新初始化状态
        /// 
        /// 上述两种情况下，都由外界保证了恢复状态的过程不会出现并发问题
        /// 情况1，启动时程序调用Init方法时，执行恢复状态，这个是使用Acotr模型保证的并发问题
        /// 情况2，在监听事件的过程中，流本身会保证处理器必须在上一个事件处理完成后，才开始下一个事件的处理，因此不会有并发问题
        /// </summary>
        /// <returns></returns>
        public virtual async ValueTask RecoverySnapshot()
        {
            // 如果快照为空，则从DB中查询快照
            // 程序启动时，快照是空
            if (this.Snapshot == null)
            {
                await this.ReadSnapshotAsync();
            }

            // 向响应流中写入DB中的事件
            await CreateGetEventFromVersion().ForEachAsync(async eventUnit => await this.ApplyEventUnit(eventUnit));

            this.AfterRecoverySnapshot();

            // If the minimum snapshot save interval version number is met, the snapshot is saved once
            if (this.Snapshot.Meta.Version - this.ActivateSnapshotVersion >= this.VertexOptions.MinSnapshotVersionInterval)
            {
                await this.SaveSnapshotAsync();
            }

            // 创建DB事件流
            IObservable<EventUnit<TPrimaryKey>> CreateGetEventFromVersion()
            {
                var observable = Observable.Create<IList<EventUnit<TPrimaryKey>>>(async observer =>
                {
                    while (true)
                    {
                        var documents = await this.EventStorage.GetList(this.ActorId, this.Snapshot.Meta.Version + 1, this.Snapshot.Meta.Version + this.VertexOptions.EventPageSize);

                        if (documents.Count == 0)
                        {
                            break;
                        }

                        var eventList = this.Convert(documents);

                        observer.OnNext(eventList);

                        // 后面没有了，则跳出
                        if (eventList.Count < this.VertexOptions.EventPageSize)
                        {
                            break;
                        }
                    }
                    observer.OnCompleted();
                    return Disposable.Empty;
                })
                  .SelectMany(x => x);

                return observable;
            }
        }

        /// <summary>
        /// 查询指定时间戳之前的所有时间
        /// </summary>
        /// <param name="endTimestamp">时间戳</param>
        /// <returns></returns>
        IObservable<EventUnit<TPrimaryKey>> CreateGetEventFromTimestamp(long endTimestamp)
        {
            var observable = Observable.Create<IList<EventUnit<TPrimaryKey>>>(async observer =>
            {
                int start = 0;
                int pageCount = this.VertexOptions.EventPageSize;
                while (true)
                {
                    var documents = await this.EventStorage.GetList(this.ActorId, endTimestamp, start, pageCount);
                    start += pageCount;

                    if (documents.Count == 0)
                    {
                        break;
                    }

                    var eventList = this.Convert(documents);

                    observer.OnNext(eventList);

                    // 后面没有了，则跳出
                    if (eventList.Count < pageCount)
                    {
                        break;
                    }
                }
                observer.OnCompleted();
                return Disposable.Empty;
            })
               .SelectMany(x => x);

            return observable;
        }

        /// <summary>
        /// 恢复完快照执行 此方法，此方法 一般用于前端 在恢复完状态后，给 对象设置PropertyManager
        /// </summary>
        protected abstract void AfterRecoverySnapshot();

        private async ValueTask ApplyEventUnit(EventUnit<TPrimaryKey> unit)
        {
            if (this.Snapshot.Meta.TryIncrementDoingVersion(this.ActorType))
            {
                this.SnapshotHandler.Apply(this.Snapshot, unit);
                this.Snapshot.Meta.UpdateVersion(unit.Meta, this.ActorType); // Version of the update process
            }
            else
            {
                // 如果中间流断了，可能会出现快照版本与事件版本对不上的问题
                // 因此重新恢复快照到事件的最新版本
                await this.RecoverySnapshot();
            }
        }

        /// <summary>
        /// 从DB读取快照,或创建空快照
        /// </summary>
        /// <returns></returns>
        protected virtual async Task ReadSnapshotAsync()
        {
            // Restore state from snapshot
            this.Snapshot = await this.SnapshotStorage.Get<T>(this.ActorId);
            if (this.Snapshot is null)
            {
                // New status
                await this.CreateSnapshot();
            }
            this.ActivateSnapshotVersion = this.Snapshot.Meta.Version;
        }

        public async ValueTask SaveSnapshotAsync(bool force = false, bool isLatest = false)
        {
            if (this.Snapshot.Meta.Version != this.Snapshot.Meta.DoingVersion)
            {
                throw new SnapshotException(this.Snapshot.Meta.ActorId.ToString(), this.ActorType, this.Snapshot.Meta.DoingVersion, this.Snapshot.Meta.Version);
            }

            // Update the snapshot if the version number difference exceeds the setting
            if ((force && (this.Snapshot.Meta.Version > this.ActivateSnapshotVersion || this.Snapshot.Meta.IsLatest != isLatest)) ||
                (this.Snapshot.Meta.Version - this.ActivateSnapshotVersion >= this.VertexOptions.SnapshotVersionInterval))
            {
                this.Snapshot.Meta.IsLatest = isLatest;
                if (this.ActivateSnapshotVersion == 0)
                {
                    await this.SnapshotStorage.Insert(this.Snapshot);
                }
                else
                {
                    await this.SnapshotStorage.Update(this.Snapshot);
                }
                this.ActivateSnapshotVersion = this.Snapshot.Meta.Version;
            }
        }

        /// <summary>
        /// 解析事件，并应用事件
        /// </summary>
        /// <param name="event">事件数据</param>
        /// <returns></returns>
        public async ValueTask ApplyEvent(byte[] @event)
        {
            if (this.TryConvertToEventUnit(@event, out EventUnit<TPrimaryKey> eventUnit))
            {
                await this.ApplyEventUnit(eventUnit);

                this.EventSubject.OnNext(eventUnit.Event);
            }
            else
            {
                throw new Exception("StreamEvent反序列化失败");
            }
        }

        public Task OnCompletedAsync()
        {
            this.EventSubject.OnCompleted();
            return Task.CompletedTask;
        }

        public Task OnErrorAsync(Exception ex)
        {
            this.EventSubject.OnError(ex);
            return Task.CompletedTask;
        }


        protected bool TryConvertToEventUnit(byte[] bytes, out EventUnit<TPrimaryKey> eventUnit)
        {
            if (EventConverter.TryParseWithNoId(bytes, out var transport) &&
                   this.EventTypeContainer.TryGet(transport.EventName, out var type))
            {
                var data = this.Serializer.Deserialize(transport.EventBytes, type);
                if (data is IEvent @event)
                {
                    var eventMeta = transport.MetaBytes.ParseToEventMeta();
                    eventUnit = new EventUnit<TPrimaryKey>
                    {
                        ActorId = this.ActorId,
                        Meta = eventMeta,
                        Event = @event
                    };
                    return true;
                }
            }
            eventUnit = default;
            return false;
        }

        /// <summary>
        /// 一定是先恢复好状态后，在监听流中的事件
        /// </summary>
        /// <returns></returns>
        public async ValueTask Init()
        {
            await this.DependencyInjection();
            await this.RecoverySnapshot();
        }

        /// <summary>
        /// TODO:这个不知道会不会有顺序问题
        /// 
        /// 从Orleans的文档上看，SMS只要生产者 在生产事件时 使用await，则事件就能保证顺序
        /// 
        /// 这个还需要测试
        /// </summary>
        /// <param name="item"></param>
        /// <param name="token"></param>
        /// <returns></returns>
        public async Task OnNextAsync(byte[] item, StreamSequenceToken token = null)
        {
            await this.ApplyEvent(item);
        }

        public ValueTask DisposeAsync()
        {
            return this.SaveSnapshotAsync();
        }

        public object GetState()
        {
            return this.Snapshot.Data;
        }
    }
}
